package com.catmiao.rdd.operate.project;

import org.apache.hadoop.yarn.util.SystemClock;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import scala.Serializable;
import scala.Tuple2;
import scala.Tuple4;

import java.util.Arrays;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

/**
 * @author ChengMiao
 * @title: Transfer_01_Map
 * @projectName spark_study
 * @description: TODO
 * @date 2024/11/25 16:27
 */
public class A01_HotTop10Category {

    public static void main(String[] args) throws InterruptedException {

        final SparkConf conf = new SparkConf();
        conf.setAppName("appName");
        conf.setMaster("local[*]");

        final JavaSparkContext jsc = new JavaSparkContext(conf);

        JavaRDD<String> rdd = jsc.textFile("datas/req_record/user_visit_action.txt");

        /**
         * 数据分析：搜索、点击、下单、支付
         * 2019-07-17_38_6502cdc9-cf95-4b08-8854-f03a25baa917_22_2019-07-17 00:00:28_null_-1_-1_null_null_15,1,20,6,4_15,88,75_9
         *  - 2019-07-17    日期
         *  - 38            用户Id
         *  - 6502cdc9-cf95-4b08-8854-f03a25baa917  sessionId
         *  - 22    页面id
         *  - 2019-07-17 00:00:28   动作时间点
         *  - null  搜索关键字
         *  - -1    品类id
         *  - -1    产品id
         *  - null  下单的品类Id
         *  - null  下单的产品Id
         *  - 15,1,20,6,4   支付的品类id
         *  - 15,88,75      支付的产品id
         *  - 9     城市id
         */

         // 1. 热门【点击数量、下单数量、支付数量】 品类top10
         JavaRDD<String> dataRdd = rdd.filter(line ->  line.split("_")[5].equals("null"));  // 过滤掉搜索行为


        JavaRDD<Category> flaMapRdd = dataRdd.flatMap(item -> {
            String[] split = item.split("_");
            if (!split[6].equals("-1")) {
                // 点击数据
                return Arrays.asList(new Category(split[6], 1, 0, 0)).iterator();
            } else if (!"null".equals(split[8])) {
                // 下单
                String[] arr = split[8].split(",");
                return Arrays.stream(arr).map(i -> {
                    return new Category(i, 0, 1, 0);
                }).iterator();
            } else {
                // 支付
                String[] arr = split[10].split(",");
                return Arrays.stream(arr).map(i -> {
                    return new Category(i, 0, 0, 1);
                }).iterator();
            }
        });

        JavaPairRDD<String, Category> pariRdd = flaMapRdd.mapToPair(tp -> {
            return new Tuple2<>(tp.getId(), tp);
        });

        JavaPairRDD<String, Category> result = pariRdd.reduceByKey(
                (o1, o2) -> {
                    return new Category(o1.getId(), o1.getN1() + o2.getN1(), o1.getN2() + o2.getN2(), o1.getN3()+o2.getN3());
                }
        );


        result.values().sortBy(obj->obj,true,2).take(10).forEach(System.out::println);

        jsc.close();


    }
}


class Category implements Serializable,Comparable<Category> {

    private String id;
    private Integer n1;
    private Integer n2;
    private Integer n3;

    public Category(String id, Integer n1, Integer n2, Integer n3) {
        this.id = id;
        this.n1 = n1;
        this.n2 = n2;
        this.n3 = n3;
    }

    public Category() {
    }

    @Override
    public int compareTo(Category o) {
        if(this.n1 > o.n1) {
            return -1;
        }else if(this.n1 < o.n1){
            return 1;
        }else {
            if(this.n2 > o.n2) {
                return -1;
            }else if(this.n2 < o.n2){
                return 1;
            }else {
                if(this.n3 > o.n3) {
                    return -1;
                }else if(this.n3 < o.n3){
                    return 1;
                }
                return 0;

            }
        }
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public Integer getN1() {
        return n1;
    }

    public void setN1(Integer n1) {
        this.n1 = n1;
    }

    public Integer getN2() {
        return n2;
    }

    public void setN2(Integer n2) {
        this.n2 = n2;
    }

    public Integer getN3() {
        return n3;
    }

    public void setN3(Integer n3) {
        this.n3 = n3;
    }

    @Override
    public String toString() {
        return "Category{" +
                "id='" + id + '\'' +
                ", n1=" + n1 +
                ", n2=" + n2 +
                ", n3=" + n3 +
                '}';
    }
}





